MINIFICPP-2603 Add Record Reader and Record Writer properties to MQTT processors #2004
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
8 tasks
There was a problem hiding this comment.
Pull Request Overview
This PR adds Record Reader and Record Writer properties to the MQTT processors (PublishMQTT and ConsumeMQTT) to enable structured record processing. Instead of treating MQTT messages as raw payloads, these processors can now parse messages into records using a Record Reader and serialize them back using a Record Writer.
- Adds Record Reader and Record Writer properties to both PublishMQTT and ConsumeMQTT processors
- Implements record-based processing logic that splits individual records for publishing and aggregates records for consuming
- Includes comprehensive test coverage for the new functionality
Reviewed Changes
Copilot reviewed 22 out of 22 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| minifi-api/include/minifi-cpp/core/Record.h | Adds constructor and contains() method to Record class for better record manipulation |
| extensions/standard-processors/processors/SplitRecord.cpp | Refactors to use common utility function for getting RecordSetIO services |
| extensions/mqtt/processors/PublishMQTT.h | Adds Record Reader/Writer properties and makes sendMessage method virtual for testing |
| extensions/mqtt/processors/PublishMQTT.cpp | Implements record processing logic that splits records into individual flow files |
| extensions/mqtt/processors/ConsumeMQTT.h | Adds Record Reader/Writer properties and Add Attributes As Fields option |
| extensions/mqtt/processors/ConsumeMQTT.cpp | Implements record aggregation logic and attribute handling for record mode |
| extensions/mqtt/processors/AbstractMQTTProcessor.h | Adds record set reader/writer member variables and makes initializeClient virtual |
| extension-utils/include/utils/ProcessorConfigUtils.h | Adds utility function for getting and validating RecordSetIO services |
| extensions/mqtt/tests/PublishMQTTTests.cpp | Comprehensive test coverage for record-based publishing functionality |
| extensions/mqtt/tests/ConsumeMQTTTests.cpp | Comprehensive test coverage for record-based consuming functionality |
| docker/test/integration/ | Integration test setup and scenarios for MQTT record processing |
| PROCESSORS.md | Documentation updates for new properties and functionality |
Comments suppressed due to low confidence (1)
extensions/mqtt/tests/ConsumeMQTTTests.cpp:151
- Spelling error: 'non-existant' should be 'non-existent'.
REQUIRE_NOTHROW(test_controller_.plan->scheduleProcessor(consume_mqtt_processor_));
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
You can also share your feedback on Copilot code review for a chance to win a $100 gift card. Take the survey.
lordgamez
force-pushed
the
MINIFICPP-2596
branch
from
27903b9 to
b7b0564
Compare
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
97ba156 to
2aae8e8
Compare
lordgamez
force-pushed
the
MINIFICPP-2596
branch
from
b7b0564 to
2d9cdcd
Compare
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
2aae8e8 to
a5d42f6
Compare
lordgamez
force-pushed
the
MINIFICPP-2596
branch
from
2d9cdcd to
e230c2c
Compare
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
a5d42f6 to
fa66a5d
Compare
lordgamez
force-pushed
the
MINIFICPP-2596
branch
from
e230c2c to
d070bdc
Compare
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
33a23ce to
295c2b7
Compare
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
295c2b7 to
14f9f1a
Compare
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
14f9f1a to
071704c
Compare
fgerlits
reviewed
Sep 29, 2025
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
071704c to
12ae77b
Compare
fgerlits
approved these changes
Oct 6, 2025
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
e05ba80 to
034d4d1
Compare
martinzink
reviewed
Oct 20, 2025
lordgamez
force-pushed
the
MINIFICPP-2603
branch
from
bf24a9e to
06aedaf
Compare
martinzink
approved these changes
Oct 20, 2025
szaszm
approved these changes
Oct 22, 2025
martinzink
pushed a commit
to martinzink/nifi-minifi-cpp
that referenced
this pull request
Nov 6, 2025
… processors Signed-off-by: Ferenc Gerlits <[email protected]> Closes apache#2004
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
https://issues.apache.org/jira/browse/MINIFICPP-2603
Depends on #1997
Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
For all changes:
Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically main)?
Is your initial contribution a single, squashed commit?
For code changes:
For documentation related changes:
Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.